package com.mercury.kafka.base;

import com.mercury.kafka.constant.RedisConstant;
import com.mercury.kafka.domain.KafKaMqMessage;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.support.Acknowledgment;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author mercury
 */
public abstract class BaseListener {

    protected static Logger logger = LoggerFactory.getLogger(BaseListener.class);


    @Resource
    private RedisTemplate<String, String> redisTemplate;


    /**
     * 业务执行方法
     *
     * @param content 消息体
     */
    protected abstract void execute(ConsumerRecord<String, KafKaMqMessage> content);

    /**
     * 失败执行
     *
     * @param content 消息体
     */
    protected abstract void failExecute(ConsumerRecord<String, KafKaMqMessage> content);


    /**
     * 处理接收到的消息
     *
     * @param record         消息
     * @param acknowledgment ack
     */
    protected void receiveMessage(ConsumerRecord<String, KafKaMqMessage> record, Acknowledgment acknowledgment) {
        try {
            if (canConsume(record)) {
                logger.info("[{}] 已经消费过", record.topic());
                acknowledgment.acknowledge();
                return;
            }
            execute(record);
            logger.info("[{}] 消费成功", record.topic());
            acknowledgment.acknowledge();
        } catch (Exception e) {
            try {
                //消费失败的处理  入库短信或者邮件之类的处理
                failExecute(record);
            } catch (Exception ex) {
                acknowledgment.nack(1000L);
            }
        }
    }


    /**
     * 是否能消费，防止重复消费
     *
     * @param record record
     * @return 结果
     */
    private Boolean canConsume(ConsumerRecord<String, KafKaMqMessage> record) {
        if (redisTemplate.opsForValue().get(RedisConstant.queueKey(record.topic(), record.value().getId())) == null) {
            return false;
        } else {
            //存储消费标志
            redisTemplate.opsForValue()
                    .set(RedisConstant.queueKey(record.topic(), record.value().getId()), "lock", 30, TimeUnit.SECONDS);
            return true;
        }
    }


}
